
#include <vector>
#include "../iface/debug.h"
#include "../iface/testerror.h"
#include "../utils/Hash.h"
#include "../utils/StringUtils.h"
#include <apr_strings.h>
#include "../aprtools/APRFile.h"
#include "../resolver/LocalResolver.h"
#include "Broker5Server.h"

#include "../db/Actor.h"
#include "../db/DataStore.h"
#include "../db/WorkInstance.h"
#include "../ssh/SecurityChain.h"
#include "../ssh/SecurityTokenKey.h"
#include "../ssh/SecurityTokenPassword.h"
#include "../ssh/SecurityTokenInteractive.h"
#include "../ssh/SSHTelescope.h"
#include "../ssh/RemoteSSHKeyFile.h"
#include "../ssh/SSHClientWrapper.h"

#include "acquire_config.h"

#define MAX_HEADERS                   10
#define LIMIT_OF_ANY_MESSAGE     2097152

#define MAX_ENCRYPTED_HEADERS        200
#define MAX_ENCRYPTED_HEADER_LENGTH 8192

/** Divine meaning from the incoming packet. This means that the
 * triggering of side effects in the broker state happens here. A
 * BrokerMessage comes in, some side effects might happen, and an
 * outgoing BrokerMessage is written.
 * 
 * @param msgin Message coming in read from network.
 * 
 * @param msgout Message that is written, goes out on the network.
 * 
 * @param i The thread number (and so Broker5Server::pipearray slot
 * number).
 * 
 * @param t_mp Memory pool to use for allocations.
 * 
 * @return Zero on success, non-zero on a problem. */
int Broker5Server::divineMeaning(BrokerMessage &msgin, BrokerMessage &msgout, int i, apr_pool_t *t_mp)
{
   char timebuf[APR_RFC822_DATE_LEN + 1];
   // returns 0 if meaning is obtained with a valid response so should
   // never return 1, but might do if error
   FATAL_IF_NULL(1, msgin.getSender(), "sender");
   FATAL_IF_NULL(1, msgin.getRecipient(), "recipient");
   FATAL_IF_NULL(1, msgin.getType(), "type");
   FATAL_IF_NULL(1, msgin.getProcessee(), "processee");
   FATAL_IF_NULL(1, msgin.getHeaderTable(), "header table");
   const APRTable *apr = msgin.getHeaderTable();
   msgout.replyTo(msgin);
   if (msgin.isType("REGISTER_USER_KEY"))
   {
      if (checkNewUser(msgin))
      {
         //msgin.obtainKeyFromReceived(0, NULL, NULL, Key::KEYPURPOSE_ENCRYPTION, msgin.getSender(), false, msgin.getSender(), dbs->keydb);
         msgout.setRecipient(msgin.getProcessee());
         //msgin.print();
         if (!(dbs->keydb->encryptionTable()->testKeyExists(msgin.getProcessee())))
         {
            const char toremove[] = { KeyDB::KEYMARK_CHAINROOT, KeyDB::KEYMARK_CHAINLATEST, '\0' };
            const char toadd[] = { KeyDB::KEYMARK_EXPENDABLE, KeyDB::KEYMARK_LIFE, KeyDB::KEYMARK_LIFE, KeyDB::KEYMARK_LIFE, '\0' };
            char *sshname = StringUtils::varprintf("%s_ssh", msgin.getProcessee());
            int keys_existed = msgin.obtainAvailableKeysFromReceived(0, toadd, toremove, msgin.getProcessee(), false, msgin.getProcessee(), dbs->keydb);
            int keys_existedssh = msgin.obtainSSHKeysFromReceived(1, "", toremove, sshname, true, sshname, dbs->keydb);
            if ((keys_existed) || (keys_existedssh))
            {
               errprint("error: inconsistent key state for registration processee '%s'", msgin.getProcessee());
               msgout.addTypeHeaderNOCRLF("REGISTRATION_USER_KEY", "INVALID_NAME");
            } else
            {
               msgout.addTypeHeaderNOCRLF("REGISTRATION_USER_KEY", "ACCEPTED");
            }
            free(sshname);
         } else
         {
            AbstractKey *enckey_new = msgin.obtainAbstractKeyFromReceived(0, Key::KEYPURPOSE_ENCRYPTION, dbs->keydb);
            Key t_key(dbs->keydb, msgin.getProcessee(), Key::KEYPURPOSE_ENCRYPTION);
            AbstractKey *enckey_exists = (AbstractKey *)Key::getEncryptionKey(dbs->keydb, NULL, msgin.getProcessee());
            int issame = enckey_new->isSameAs(*enckey_exists);
            if (issame)
               msgout.addTypeHeaderNOCRLF("REGISTRATION_USER_KEY", "AUTHORISED");
            else
               msgout.addTypeHeaderNOCRLF("REGISTRATION_USER_KEY", "INVALID_NAME");
            delete enckey_new;
            delete enckey_exists;
         }
         return 0;
      } else
      {
         msgout.addTypeHeaderNOCRLF("REGISTRATION_USER_KEY", "DENIED");
         return 0;
      }
   }
   if (checkNewUser(msgin))
   {
      errprint("new user with no permission using an unauthorised message type");
      msgout.addTypeHeaderNOCRLF("DENIED", "UNAUTHORISED");
      return 1;
   }
   if (!checkPermissionsOkay(msgin))
   {
      errprint("permission for '%s' to perform action '%s' on '%s' denied!", msgin.getSender(), msgin.getType(), msgin.getProcessee());
      msgout.addTypeHeaderNOCRLF("DENIED", "NO_PERMISSION");
      return 1;
   }
   if (msgin.isType("REGISTER_SSH_LOGIN"))
   {
      msgout.setRecipient(msgin.getProcessee());
      //msgin.print();
      char *sshname = StringUtils::varprintf("%s_ssh", msgin.getProcessee());
      // construct an SSH authentication chain
      SecurityChain *secur_chain = new SecurityChain();
      if (msgin.getHowManyPayloads() > 0)
      {
         char *temp_sshkeyname = StringUtils::varprintf(".temp_%s", sshname);
         const char keymarks_add[] = { KeyDB::KEYMARK_EXPENDABLE, '\0' };
         const char keymarks_rmv[] = { KeyDB::KEYMARK_PERMANENT, '\0' };
         msgin.obtainAllKeysFromReceived(0, keymarks_add, keymarks_rmv, sshname, false, temp_sshkeyname, dbs->keydb);
         SecurityTokenKey *secur_tok_key = new SecurityTokenKey(temp_sshkeyname, temp_sshkeyname, NULL, dbs->keydb);
         secur_chain->addSecurityToken(secur_tok_key);
         free(temp_sshkeyname);
      }
      const APRTable *headertable = msgin.getHeaderTable();
      const char *t_ssh_password = headertable->get("ssh_password");
      const char *t_unlock_password = headertable->get("unlock_password");
      if (t_ssh_password != NULL)
      {
         SecurityTokenPassword *secur_tok_pass = new SecurityTokenPassword(t_ssh_password);
         secur_chain->addSecurityToken(secur_tok_pass);
         SecurityTokenInteractive *secur_tok_intr = new SecurityTokenInteractive(t_ssh_password);
         secur_chain->addSecurityToken(secur_tok_intr);
      }
      // watch out! sshname != msgin.getProcessee()
      // also make sure ssh_port is sent by the client!
      const char *remote_port_string = headertable->get("ssh_port");
      int remote_port_int = (remote_port_string) ? (atoi(remote_port_string)) : 22; // default SSH port 22 (but shouldn't ever use this hard-coded value)
      if (remote_port_string == NULL) errprint("warning: using built-in default SSH port");
      SSHTelescope *ssh_tele = new SSHTelescope(sshname, *(dbs->networkdb), *(dbs->keydb), headertable->get("ssh_machine"), remote_port_int, mp, t_unlock_password);
      SSHClient *finalclient = NULL;
      if (ssh_tele->telescopeOkay())
      {
         int my_local_port = ssh_tele->getLocalPort();
         finalclient = new SSHClient(
            (my_local_port) ? "127.0.0.1" : headertable->get("ssh_machine"),
            (my_local_port) ? (my_local_port) : remote_port_int,
            mp, NULL, 0,
            headertable->get("ssh_machine"), remote_port_int, dbs->pendingdb);
         if (finalclient->useSecurityChain(secur_chain, headertable->get("ssh_username")))
         {
            errprint("SSH authentication failed for final client");
            msgout.addTypeHeaderNOCRLF("SSH_LOGIN_STATUS", "INVALID_AUTHENTICATION"); // FINGERPRINT_ERROR also needs adding!
         } else
         {
            // yay! now we grab a RemoteSSHKeyFile and impose our key
            // structure onto the remote host
            errprint("SSH client opened successfully");
            SSHClientWrapper::doRemoteKeyUpdates(finalclient, dbs->networkdb, dbs->keydb, APP_KEY_TAG,
               sshname, headertable->get("ssh_username"), headertable->get("ssh_machine"), t_unlock_password);
            // Adds the host if it is unknown
            dbs->networkdb->addHost(headertable->get("ssh_machine"), remote_port_int, NULL, NULL);
            msgout.addTypeHeaderNOCRLF("SSH_LOGIN_STATUS", "SUCCESS");
            if (!(dbs->actorsdb->testActorExists(msgin.getSender())))
            {
               char *new_username_copy = 
                  dbs->actorsdb->addUser(msgin.getSender(), NULL, 0, ((int64_t)1048576)*((int64_t)1048576), 3600);
               free(new_username_copy);
            }
         }
         delete finalclient;
      } else
      {
         msgout.addTypeHeaderNOCRLF("SSH_LOGIN_STATUS", "NO_ACCESS");
      }
      if (msgin.getHowManyPayloads() > 0)
      {
         char *temp_sshkeyname = StringUtils::varprintf(".temp_%s", sshname);
         dbs->keydb->deleteAllKeys(temp_sshkeyname);
         free(temp_sshkeyname);
      }
      delete secur_chain;
      delete ssh_tele;
      free(sshname);
      return 0;
   }
   if (msgin.isType("START_DATA_STORE"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      const APRTable *headertable = msgin.getHeaderTable();
      if ((headertable->get("Data-Store-ID")) && (headertable->get("Data-Store-Decryption-Key")))
      {
         DataStore dstore(dbs->actorsdb, headertable->get("Data-Store-ID"), msgin.getSender());
         DataStore::datastorestate dstate =
            dstore.startNew(msgin, dbs->keydb, headertable->get("Data-Store-Decryption-Key"), t_mp);
         const char *error_msg = dstore.stateToErrorMessage(dstate);
         const char *packet_msg = dstore.stateToMessageTag(dstate);
         if ((dstate != DataStore::FILE_CREATION_SUCCEEDED) && (dstate != DataStore::ACTOR_CREATION_SUCCEEDED))
         {
            errprint("data store '%s' creation: %s", headertable->get("Data-Store-ID"), error_msg);
            msgout.addTypeHeaderNOCRLF("DATA_STORE_DENIED", packet_msg);
            dstore.setUncommitable();
         } else
         {
            msgout.addTypeHeaderNOCRLF("DATA_STORE_CREATED", "SUCCESS");
            dstore.update();
         }
      }
      return 0;
   }
   if (msgin.isType("RESOLVE_KEY_EQUIVALENCE"))
   {
      msgout.setRecipient(msgin.getSender());
      char *new_key_name = dbs->actorsdb->getEquivalentKey(msgin.getProcessee());
      if (new_key_name)
      {
         msgout.addTypeHeaderNOCRLF("ALTERNATIVE_KEY", new_key_name);
         free(new_key_name);
      } else
      {
         msgout.addTypeHeaderNOCRLF("NO_EQUIVALENT", "NONE");
      }
      //msgout.print();
      return 0;
   }
   if (!(dbs->actorsdb->testActorExists(msgin.getProcessee())))
   {
      errprint("user '%s' does not have credentials", msgin.getProcessee());
      msgout.addTypeHeaderNOCRLF("DENIED", "BAD_CREDENTIALS");
      return 1;
   }
   // by this point, the actor must exist.
   // you can't create a data store without existing, so if you don't exist,
   // all the manipulation of data hereon in is inaccessible to you.
   if (msgin.isType("CHECK_DATA_BLOCK"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      DataStore dstore(dbs->actorsdb, msgin.getProcessee(), msgin.getSender());
      DataStore::datastorestate dstate = dstore.checkBlockStatus(msgin.getHeaderTable(), t_mp);
      msgout.addTypeHeaderNOCRLF("DATA_BLOCK_STATUS", dstore.stateToMessageTag(dstate));
      return 0;
   }
   if (msgin.isType("CHECK_DATA_STORE"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      DataStore dstore(dbs->actorsdb, msgin.getProcessee(), msgin.getSender());
      // Send data on the actor state!
      dstore.writeOutStatus(msgout, t_mp);
      return 0;
   }
   if (msgin.isType("PURPOSE_DATASTORE"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      DataStore dstore(dbs->actorsdb, msgin.getProcessee(), msgin.getSender());
      const APRTable *headertable = msgin.getHeaderTable();
      int changed_complete = 0;
      int changed_work = 0;
      if (headertable->get("Complete"))
      {
         changed_complete = !(dstore.checkActorMark(ActorsDB::ACTORMARK_DATA_COMPLETE));
         dstore.addActorMark(ActorsDB::ACTORMARK_DATA_COMPLETE, 1);
      }
      if (headertable->get("Work"))
      {
         changed_work = !(dstore.checkActorMark(ActorsDB::ACTORMARK_DATA_IS_WORK));
         dstore.addActorMark(ActorsDB::ACTORMARK_DATA_IS_WORK, 1);
      }
      if (headertable->get("Results"))
         dstore.addActorMark(ActorsDB::ACTORMARK_DATA_IS_RESULTS, 1);
      dstore.update();
      if (((changed_complete) || (changed_work)) && (dstore.checkActorMark(ActorsDB::ACTORMARK_DATA_IS_WORK)))
         if (headertable->get("Work"))
         {
            int retval = WorkInstance::instantiateWork(msgin, dbs, t_mp);
            if (retval != 0)
            {
               // set as incomplete, because it didn't actually complete
               dstore.removeActorMark(ActorsDB::ACTORMARK_DATA_COMPLETE);
            }
         }
      // Send data on the actor state!
      dstore.writeOutStatus(msgout, t_mp);
      return 0;
   }
   if (msgin.isType("UPLOAD_DATA_BLOCK"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      DataStore dstore(dbs->actorsdb, msgin.getProcessee(), msgin.getSender());
      DataStore::datastorestate dstate = dstore.checkBlockStatus(msgin.getHeaderTable(), t_mp);
      // If it is incomplete then we supplement the file with the new data
      if (dstate == DataStore::DATA_BLOCK_INCOMPLETE)
      {
         dstate = dstore.writeBlock(msgin, t_mp);
         if (dstate != DataStore::TRANSFER_SUCCEEDED)
         {
            msgout.addTypeHeaderNOCRLF("DATA_UPLOAD_ERROR", dstore.stateToMessageTag(dstate));
         } else
         {
            int data_blocks_done = 0;
            dbs->actorsdb->sqlGetSingleIntf(&data_blocks_done, "select blocksdone from 'actors' where name='%s';", msgin.getProcessee());
            data_blocks_done++;
            dbs->actorsdb->sqlInstantExecf("update 'actors' set blocksdone='%d' where name='%s';", data_blocks_done, msgin.getProcessee());
            msgout.addTypeHeaderNOCRLF("DATA_BLOCK_STATUS", "DATA_BLOCK_COMPLETE");
         }
      } else
         msgout.addTypeHeaderNOCRLF("DATA_BLOCK_STATUS", dstore.stateToMessageTag(dstate));
      return 0;
   }
   if (msgin.isType("REQUEST_WORK"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      WorkInstance winsta(dbs, msgin.getProcessee(), msgin.getSender());
      msgout.addTypeHeaderNOCRLF("WORK_INSTANCE_INFO", msgin.getProcessee());
      msgout.addHeaderLineNOCRLF("DS-Name: %s", winsta.getWorkInstanceDataStore());
      msgout.addHeaderLineNOCRLF("Local-Instance: %d", winsta.getWorkInstanceLocalIDAsInt());
      return 0;
   }
   if (msgin.isType("GET_DATA_BLOCK"))
   {
      msgout.setRecipient(msgin.getSender());
      //msgin.print();
      const APRTable *headertable = msgin.getHeaderTable();
      if (headertable->get("Block-ID") == NULL)
      {
         msgout.addTypeHeaderNOCRLF("DOWNLOAD_DATA_ERROR", "BLOCK_ID_MISSING");
         return 0;
      }
      DataStore dstore(dbs->actorsdb, msgin.getProcessee(), msgin.getSender());
      dstore.readBlock(msgout, atoi(headertable->get("Block-ID")), t_mp);
      //msgout.print();
      return 0;
   }
   if (msgin.isType("RESOLVE_PARENT"))
   {
      msgout.setRecipient(msgin.getSender());
      Resolver *server_resolver = (Resolver *)(new LocalResolver(dbs->actorsdb, t_mp));
      const APRTable *headertable = msgin.getHeaderTable();
      // need to check for NULL      
      if (strstr(headertable->get("Hierarchical-Namespace"), msgin.getProcessee()) != NULL)
      {
         const char *server_response = NULL;
         if (server_response = server_resolver->get(headertable->get("Hierarchical-Parent")))
         {
            const char *parent_id = server_response;
            const char *parent_ext_name = NULL;
            if (parent_id != NULL)
            {
               Actor p_actor(dbs->actorsdb);
               p_actor.setName(parent_id);
               parent_ext_name = p_actor.getExternalAlias();
            }
            if ((parent_id) && (strlen(parent_id) > 0))
            {
               msgout.addTypeHeaderNOCRLF("PARENT_DATA", "SUCCESS");
               msgout.addVerbatimPayloadString(parent_id);
               msgout.addVerbatimPayloadString(parent_ext_name);
            } else
            {
               msgout.addTypeHeaderNOCRLF("NO_PARENT", "SUCCESS");
            }
         } else msgout.addTypeHeaderNOCRLF("GET_PARENT_ERROR", "ERROR");
      }
      return 0;
   }
   if (msgin.isType("RESOLVE_CHILDREN"))
   {
      msgout.setRecipient(msgin.getSender());
      Resolver *server_resolver = (Resolver *)(new LocalResolver(dbs->actorsdb, t_mp));
      const APRTable *headertable = msgin.getHeaderTable();
      // need to check for NULL      
      if (strstr(headertable->get("Hierarchical-Namespace"), msgin.getProcessee()) != NULL)
      {
         const char *server_response = NULL;
         if (server_response = server_resolver->get(headertable->get("Hierarchical-Parent")))
         {
            std::vector<char *> children_ids;
            std::vector<char *> children_names;
            int retcode = dbs->actorsdb->getChildren(&children_ids, server_response);
            int retcode2 = dbs->actorsdb->getExternalAliasesOfChildren(&children_names, server_response);
            if ((retcode == 0) && (retcode2 == 0))
            {
               if ((children_names.size() > 0) && (children_ids.size() > 0))
               {
                  char *ids_list = StringUtils::flattenStringList(&children_ids, ",");
                  char *names_list = StringUtils::flattenStringList(&children_names, ",");
                  msgout.addTypeHeaderNOCRLF("CHILDREN_DATA", "SUCCESS");
                  msgout.addVerbatimPayloadString(ids_list);
                  msgout.addVerbatimPayloadString(names_list);
                  free(ids_list);
                  free(names_list);
               } else
               {
                  msgout.addTypeHeaderNOCRLF("NO_CHILDREN", "SUCCESS");
               }
            } else msgout.addTypeHeaderNOCRLF("GET_CHILDREN_ERROR", "ERROR");
            StringUtils::destroyStringList(children_ids);
            StringUtils::destroyStringList(children_names);
         }
      }
      delete server_resolver;
      return 0;
   }
   if (msgin.isType("HAPPY_COMPUTE"))
   {
      msgout.setRecipient(msgin.getSender());
      msgout.addTypeHeaderNOCRLF("HAPPY_BROKER", msgin.getProcessee());
      return 0;
   }
   if ((msgin.isType("RESOLVE_NAME")) || (msgin.isType("RESOLVE_METADATA")))
   {
      msgout.setRecipient(msgin.getSender());
      Resolver *server_resolver = (Resolver *)(new LocalResolver(dbs->actorsdb, t_mp));
      const APRTable *headertable = msgin.getHeaderTable();
      // need to check for NULL      
      if (strstr(headertable->get("Hierarchical-Namespace"), msgin.getProcessee()) != NULL)
      {
         const char *server_response = NULL;
         if (server_response = server_resolver->get(headertable->get("Hierarchical-Parent")))
         {
            if (msgin.isType("RESOLVE_METADATA"))
            {
               msgout.addTypeHeaderNOCRLF("RESOLUTION", "SUCCESS");
               msgout.addHeaderLineNOCRLF("Real-ID: %s", server_response);
               Actor tactor(dbs->actorsdb);
               tactor.setName(server_response);
               APRTable *ttable = tactor.getMetadata(t_mp);
               std::vector<char *> *ttable_output_headers = ttable->getAllAsStringList(": ", "");
               for (int i = 0; i < ttable_output_headers->size(); i++)
                  msgout.addHeaderLineNOCRLF("%s", ttable_output_headers->at(i));
               StringUtils::destroyStringList(ttable_output_headers);
               delete ttable;
            } else
            {
               msgout.addTypeHeaderNOCRLF("RESOLUTION_METADATA", "FAILED");
            }
         } else if (server_response = server_resolver->get(headertable->get("Hierarchical-Namespace")))
         {
            if (msgin.isType("RESOLVE_NAME"))
            {
               msgout.addTypeHeaderNOCRLF("RESOLUTION", "SUCCESS");
               msgout.addHeaderLineNOCRLF("Real-ID: %s", server_response);
               const char *type_request = StringUtils::mpprintf(t_mp, "%s.type", headertable->get("Hierarchical-Namespace"));
               const char *server_type_response = server_resolver->get(type_request);
               if (server_type_response)
                  msgout.addHeaderLineNOCRLF("Real-Type: %s", server_type_response);
            } else
            {
               msgout.addTypeHeaderNOCRLF("RESOLUTION_NAME", "FAILED");
            }
         } else
         {
            msgout.addTypeHeaderNOCRLF("RESOLUTION", "FAILED");
         }
      } else
      {
         msgout.addTypeHeaderNOCRLF("RESOLUTION_DENIED", "ERROR");
      }
      delete server_resolver;
      return 0;
   }
   if (msgin.isType("RESOLVE_PROGRESS"))
   {
      msgout.setRecipient(msgin.getSender());
      Resolver *server_resolver = (Resolver *)(new LocalResolver(dbs->actorsdb, t_mp));
      const APRTable *headertable = msgin.getHeaderTable();
      // need to check for NULL
      if (strstr(headertable->get("Hierarchical-Namespace"), msgin.getProcessee()) != NULL)
      {
         const char *server_response = server_resolver->get(headertable->get("Hierarchical-Namespace"));
         if (server_response)
         {
            msgout.addTypeHeaderNOCRLF("RESOLUTION", "SUCCESS");
            msgout.addHeaderLineNOCRLF("Real-ID: %s", server_response);
            const char *type_request = StringUtils::mpprintf(t_mp, "%s.progress", headertable->get("Hierarchical-Namespace"));
            const char *server_type_response = server_resolver->get(type_request);
            if (server_type_response)
            {
               msgout.addHeaderLineNOCRLF("Real-Progress: %s", server_type_response);
            }
         } else
         {
            msgout.addTypeHeaderNOCRLF("RESOLUTION", "FAILED");
         }
      } else
      {
         msgout.addTypeHeaderNOCRLF("RESOLUTION_DENIED", "ERROR");
      }
      delete server_resolver;
      return 0;
   }
   if (msgin.isType("UPDATE_BLOCKS"))
   {
      msgout.setRecipient(msgin.getSender());
      Actor tactor(dbs->actorsdb);
      tactor.setName(msgin.getProcessee());
      if (tactor.updateBlocks(msgin))
      {
         msgout.addTypeHeaderNOCRLF("BLOCK_UPDATE", "SUCCESS");
         tactor.update();
      } else
      {
         msgout.addTypeHeaderNOCRLF("BLOCK_UPDATE", "FAILED");
         errprint("actor block count update failed");
         msgin.print();
      }
      return 0;
   }
   msgout.addTypeHeaderNOCRLF("DENIED", "UNKNOWN");
   msgin.print();
   return 1;
}

/** Check if the key name has a new user indicator on it. If so, it is
 * saved to the database, but will be eliminated if it is not made
 * concrete by further credentials.
 * 
 * @param msgin Incoming message whose sender is to be checked.
 * 
 * @return One if the sender is a new user, zero if it is not new. */
int Broker5Server::checkNewUser(BrokerMessage &msgin)
{
   return ((strcmp(msgin.getProcessee(), &((msgin.getSender())[1])) == 0) &&
           ((msgin.getSender())[0] == '.'));
}

/** Check if the message \link BrokerMessage::sender sender \endlink
 * has permission to modify the referenced \link
 * BrokerMessage::processee processee \endlink.
 * 
 * @param msgin The message whose permission need checking.
 * 
 * @return One if permission is granted, zero if it is denied. */
int Broker5Server::checkPermissionsOkay(BrokerMessage &msgin)
{
   if (msgin.getProcessee() == NULL) { errprint("processee is NULL!"); return 0; }
   if (msgin.getSender() == NULL) { errprint("sender is NULL?"); return 0; }
   if (dbs->actorsdb->testActorExists(msgin.getProcessee()))
   {
      return dbs->actorsdb->isEquivalentPermission(msgin.getProcessee(), msgin.getSender());
   } else
   {
      char *owner_name = NULL;
      // test the key tables for a key that matches the processee (processee might be a key)
      // find the owner of the key that matches, test permissions against that
      if (dbs->keydb->encryptionTable()->testKeyExists(msgin.getProcessee()))
         if (owner_name = dbs->keydb->encryptionTable()->getKeyOwner(msgin.getProcessee()))
         {
            int retcode = dbs->actorsdb->isEquivalentPermission(owner_name, msgin.getSender());
            free(owner_name);
            return retcode;
         }
      if (dbs->keydb->decryptionTable()->testKeyExists(msgin.getProcessee()))
         if (owner_name = dbs->keydb->decryptionTable()->getKeyOwner(msgin.getProcessee()))
         {
            int retcode = dbs->actorsdb->isEquivalentPermission(owner_name, msgin.getSender());
            free(owner_name);
            return retcode;
         }
      if (dbs->keydb->signingTable()->testKeyExists(msgin.getProcessee()))
         if (owner_name = dbs->keydb->signingTable()->getKeyOwner(msgin.getProcessee()))
         {
            int retcode = dbs->actorsdb->isEquivalentPermission(owner_name, msgin.getSender());
            free(owner_name);
            return retcode;
         }
      if (dbs->keydb->checkingTable()->testKeyExists(msgin.getProcessee()))
         if (owner_name = dbs->keydb->checkingTable()->getKeyOwner(msgin.getProcessee()))
         {
            int retcode = dbs->actorsdb->isEquivalentPermission(owner_name, msgin.getSender());
            free(owner_name);
            return retcode;
         }
   }
   // fail
   return 0;
}

/** Generate a HTML file from the pending database which acts as a
 * security console, only accessible from the local machine.
 * 
 * @param i The thread/#pipearray number to serve it onto.
 * 
 * @param t_mp The memory pool to use for allocation. */
void Broker5Server::generateConsole(int i, apr_pool_t *t_mp)
{
   static const char *html_start = "<html><head><title>Console</title></head>" CRLF_STR;
   static const char *open_form = "<body><form name=\"permissions\" action=\"console.html\" method=\"POST\">" CRLF_STR;
   static const char *open_div = "<div align=\"center\"><br />" CRLF_STR;
   static const char *individ_header = "<h2>%s</h2>" CRLF_STR;
   static const char *user_record = "'%s' at %s from %s with %s &nbsp; " CRLF_STR;
   static const char *keyup_record = "'%s' expected host fingerprint '%s', obtained '%s' &nbsp; " CRLF_STR;
   static const char *input_radio = "<input type=\"radio\" name=\"%s\" value=\"%s\" />%s" CRLF_STR;
   static const char *input_cradio = "<input type=\"radio\" name=\"%s\" value=\"%s\" checked />%s" CRLF_STR;
   static const char *next_line = "<br />" CRLF_STR;
   static const char *submit_button = "<br /><br /><input type=\"submit\" action=\"console.html\" value=\"Submit Changes\" />" CRLF_STR;
   static const char *html_stop = "</div></form></body></html>" CRLF_STR CRLF_STR;
   static const char *firstheader = "HTTP/1.0 200 OK" CRLF_STR CRLF_STR;
   pipearray[i].handler_to_send->writeToStream((char *)firstheader, strlen(firstheader));
   pipearray[i].handler_to_send->writeToStream((char *)html_start, strlen(html_start));
   pipearray[i].handler_to_send->writeToStream((char *)open_form, strlen(open_form));
   pipearray[i].handler_to_send->writeToStream((char *)open_div, strlen(open_div));
   std::vector<char *> *pendingkeyupdates = dbs->pendingdb->getAllPendingHostKeyUpdates();
#define COLUMNS_KEYUPDATE_FIELDS 3
   int countkeyupdaterows = pendingkeyupdates->size() / COLUMNS_KEYUPDATE_FIELDS;
   if (countkeyupdaterows > 0)
   {
      pipearray[i].handler_to_send->writefToStream((char *)individ_header, "Host fingerprints");
      for (int k = 0; k < countkeyupdaterows; k++)
      {
         const char *t_hostname = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0);
         const char *t_oldfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+1);
         const char *t_newfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+2);
         int needs_none = ((t_oldfp == NULL) || (strlen(t_oldfp) == 0));
         // I can make this nicer to look at later! - format the key
         // fingerprints the same!
         pipearray[i].handler_to_send->writefToStream((char *)keyup_record, t_hostname, needs_none ? t_oldfp : "(none)", t_newfp);
         char *key_fingerprint_as_variable_name = (char *)malloc(sizeof(char)*(strlen(t_newfp)+1+1));
         key_fingerprint_as_variable_name[0] = '_';
         strcpy(&(key_fingerprint_as_variable_name[1]), t_newfp);
         pipearray[i].handler_to_send->writefToStream((char *)input_radio, key_fingerprint_as_variable_name, "Allow", "Allow");
         pipearray[i].handler_to_send->writefToStream((char *)input_radio, key_fingerprint_as_variable_name, "Deny", "Deny");
         pipearray[i].handler_to_send->writefToStream((char *)input_cradio, key_fingerprint_as_variable_name, "Ignore", "Ignore");
         pipearray[i].handler_to_send->writeToStream((char *)next_line, strlen(next_line));
         free(key_fingerprint_as_variable_name);
      }
   }
   if (countkeyupdaterows > 0)
      pipearray[i].handler_to_send->writeToStream((char *)submit_button, strlen(submit_button));
   pipearray[i].handler_to_send->writeToStream((char *)html_stop, strlen(html_stop));
   pipearray[i].handler_to_send->writeEOS();
   StringUtils::destroyStringList(pendingkeyupdates);
}

/** Parse and act upon the result given by HTTP POST from the
 * administrator who has filled in the form created by the security
 * console in #generateConsole.
 * 
 * @param data The table data produced by the HTTP POST handler. */
void Broker5Server::understandPOSTData(APRTable *data)
{
   std::vector<char *> *pendingkeyupdates = dbs->pendingdb->getAllPendingHostKeyUpdates();
#define COLUMNS_KEYUPDATE_FIELDS 3
   int countkeyupdaterows = pendingkeyupdates->size() / COLUMNS_KEYUPDATE_FIELDS;
   if (countkeyupdaterows > 0)
   {
      for (int k = 0; k < countkeyupdaterows; k++)
      {
         const char *t_newfp = pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+2);
         char *key_fingerprint_as_variable_name = (char *)malloc(sizeof(char)*(strlen(t_newfp)+1+1));
         key_fingerprint_as_variable_name[0] = '_';
         strcpy(&(key_fingerprint_as_variable_name[1]), t_newfp);
         const char *tableval = data->get("%s", key_fingerprint_as_variable_name);
         if (tableval)
         {
//            if (strcmp(tableval, "Allow") == 0)
//               dbs->clustersdb->updateKeyFingerprint(t_newfp, pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0));
            if ((strcmp(tableval, "Allow") == 0) || (strcmp(tableval, "Deny") == 0))
               dbs->pendingdb->removePendingHostKeyUpdate(pendingkeyupdates->at(k*COLUMNS_KEYUPDATE_FIELDS+0));
         }
         free(key_fingerprint_as_variable_name);
      }
   }
   StringUtils::destroyStringList(pendingkeyupdates);
}

/** Serve a file if it is in the white list. This means that it is
 * dynamically generated here by us.
 * 
 * @param htable The table of incoming headers.
 * 
 * @param i The thread this function belongs to, and which pipe to
 * use.
 * 
 * @param t_mp The APR memory pool to use for any memory usage.
 * 
 * @param systemwhitelist The white list of files that are okay to
 * serve over HTTP.
 * 
 * @return True if the file was served, false if the file was not on
 * the whitelist/does not exist. */
int Broker5Server::serveStreamFromSystemList(APRTable *htable, int i, apr_pool_t *t_mp, const char **systemwhitelist)
{
   char *dstr = apr_pstrdup(t_mp, htable->get("GET"));
   char *httpver = NULL;
   char *firsttok = apr_strtok(dstr, " ", &httpver);
   int count = 0;
   int passes = 0;
   // see if it matches any in the white list
   while (systemwhitelist[count] != NULL)
      passes |= (strcmp(systemwhitelist[count++], firsttok) == 0);
   if (passes == 0) return 0;
   if (strstr(firsttok, ".htm") == NULL)
   {
      streamPlainTextKey(i, &(firsttok[1]), NULL);
      return 1;
   }
   if (strcmp(getClientAddressAsStaticString(i), "127.0.0.1") != 0)
      return 0;
   generateConsole(i, t_mp);
   return 1;
}

/** Handle and reply to a HTTP GET session request.
 * 
 * @param i The thread this function belongs to, and which pipe in the
 * #pipearray to use.
 * 
 * @param t_mp The APR memory pool to use to allocate from. */
void Broker5Server::handleGETSession(int i, apr_pool_t *t_mp)
{
   static const char *const_whitelist[] = { "bootstrap.sh", "python_bootstrap.sh",
      "queue.sh", "compute", "_acquirepy", NULL };
   static const char *systemwhitelist[] = { "/console.html", "/broker.enc", "/broker.chk", NULL };
   APRTable *headertable = getHeaders(i, t_mp);
   if (headertable == NULL)
   {
      replyWithError404(i);
      errprint("header table missing (out of memory?).");
      return;
   }
   if (headertable->isEmpty())
   {
      replyWithError404(i);
      errprint("header table empty (unknown error).");
      delete headertable;
      return;
   }
   std::vector<char *> *vec_whitelist = new std::vector<char *>();
   int co = 0;
   while (const_whitelist[co] != NULL)
   {
      vec_whitelist->push_back(StringUtils::varprintf("/hosted_files/%s", const_whitelist[co]));
      co++;
   }
   vec_whitelist->push_back(NULL);
   if (serveFilesFromWhiteList(headertable, i, t_mp, (const char **)(&(vec_whitelist->at(0)))))
   {
      StringUtils::destroyStringList(vec_whitelist);
      delete headertable;
      return;
   }
   StringUtils::destroyStringList(vec_whitelist);
   if (serveStreamFromSystemList(headertable, i, t_mp, systemwhitelist))
   {
      delete headertable;
      return;
   }
   errprint("can not match request for %s", headertable->get("GET"));
   replyWithError404(i);
   delete headertable;
}

/** Handle and reply to a HTTP POST session request.
 * 
 * @param i The thread this function belongs to, and which pipe in the
 * #pipearray to use.
 *
 * @param t_mp The APR memory pool to use to allocate from. */
void Broker5Server::handlePOSTSession(int i, apr_pool_t *t_mp)
{
   if (strcmp(getClientAddressAsStaticString(i), "127.0.0.1") != 0)
   {
      errprint("erroneous post by %s", getClientAddressAsStaticString(i));
      pipearray[i].handler_to_send->writeEOS();
      return;
   }
   APRTable *headertable = getHeaders(i, t_mp);
   apr_size_t contentlength = getContentLength(headertable);
   APRTable *posttable = getPOSTData(i, t_mp, contentlength);
   understandPOSTData(posttable);
   redirectAfterPOST(i, t_mp, "/console.html");
   delete posttable;
   delete headertable;
}

/** Handle and reply to an encrypted broker session request.
 * 
 * @param i The thread this function belongs to, and which pipe in the
 * #pipearray to use.
 *
 * @param t_mp The APR memory pool to use to allocate from. */
void Broker5Server::handleBrokerSession(int i, apr_pool_t *t_mp)
{
   BrokerMessage msg(t_mp);
   if (!(msg.receiveEncryptedOnBucketBrigade(pipearray[i].recv_to_handler, dbs->keydb, pipearray[i].waiting)))
   {
      BrokerMessage sendmsg(t_mp);
      if (divineMeaning(msg, sendmsg, i, t_mp))
      {
         errprint("message failed to be understood/authorised!");
      }
      if (!(sendmsg.sendEncryptedOnBucketBrigade(pipearray[i].handler_to_send, dbs->keydb)))
      {
         fprintf(stdout, "*");
         fflush(stdout);
         if ((sendmsg.getRecipient())[0] == '.')
         {
            dbs->keydb->encryptionTable()->deleteKey(sendmsg.getRecipient());
            dbs->keydb->checkingTable()->deleteKey(sendmsg.getRecipient());
         }
         if (sendmsg.mustWeDestroySendingKey())
         {
            dbs->keydb->decryptionTable()->deleteKey(sendmsg.getSender());
            dbs->keydb->signingTable()->deleteKey(sendmsg.getSender());
         }
      } else errprint("could not write reply");
   } else
   {
      errprint("could not read the message");
   }
   pipearray[i].handler_to_send->writeEOS();
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param threadid The integer thread (and pipe in the #pipearray) ID
 * number.
 * 
 * @param localmp The memory pool local to this thread. */
void Broker5Server::handleClientsOnThread(int threadid, apr_pool_t *localmp)
{
   pipearray[threadid].sessionstart->lock();
   while (!sessionstop)
   {
      pipearray[threadid].sessionstart->wait();
      if (sessionstop) break;
      apr_pool_t *sp;
      apr_pool_create(&sp, localmp);
      pipearray[threadid].waiting->lock();
      while ((pipearray[threadid].recv_to_handler) &&
             (pipearray[threadid].recv_to_handler->realLength() < 4) &&
             (!(pipearray[threadid].recv_to_handler->isClosed())))
         pipearray[threadid].waiting->wait();
      char fst[5];
      if ((pipearray[threadid].recv_to_handler) &&
          (pipearray[threadid].recv_to_handler->realLength() >= 4))
      {
         pipearray[threadid].recv_to_handler->readFromStream(fst, 4);
         fst[4] = '\0';
      }
      else
         fst[0] = '\0';
      if ((!(pipearray[threadid].handler_to_send->isClosed())) && 
          (!(pipearray[threadid].recv_to_handler->isClosed())))
      {
         if (strcmp(fst, "GET ") == 0)
            handleGETSession(threadid, localmp);
         else if (strcmp(fst, "POST") == 0)
            handlePOSTSession(threadid, localmp);
         else if (strcmp(fst, "FROM") == 0)
            handleBrokerSession(threadid, sp);
         else
            pipearray[threadid].handler_to_send->writeEOS();
      } else
         pipearray[threadid].handler_to_send->writeEOS();
      // Make sure these are closed so that this pipe can be reused.
      pipearray[threadid].recv_to_handler->clear();
      pipearray[threadid].recv_to_handler->writeEOS();

      pipearray[threadid].waiting->unlock();
      apr_pool_clear(sp);
      apr_pool_destroy(sp);
   }
   pipearray[threadid].sessionstart->unlock();
}

/** Start handling clients to the server (the client handler threads
 * run this).
 * 
 * @param tthread The APR pointer to this thread grabbed.
 * 
 * @param args The void pointer given to the thread. This is actually
 * a pointer to an array of two void pointers, which are the this
 * pointer and a pointer to the thread ID respectively.
 * 
 * @return Return is NULL. */
void *Broker5Server::handleClientsOnThreadStatic(apr_thread_t *tthread, void *args)
{
   apr_status_t status;
   Broker5Server *kthis = *((Broker5Server **)(((void **)(args))[0]));
   int i = *((int *)(((void **)(args))[1]));
   apr_pool_t *upperpool, *t_mp;
   upperpool = apr_thread_pool_get(tthread);
   status = apr_pool_create(&t_mp, upperpool);
   CheckAPRError(status);

   kthis->handleClientsOnThread(i, t_mp);
   
   // do not clear memory pool t_mp, as bucket brigades may be remaining
   ERR_remove_state(0);
   apr_thread_exit(tthread, APR_SUCCESS);
   return NULL;
}

/** Export the latest in the chain of the given owner to the client on
 * the given thread.
 * 
 * @param threadid The number ofthe thread and ID of the pipe in the
 * #pipearray to use.
 * 
 * @param extended_ownername An owner name of the chain whose latest
 * key is to be put into the stream.  An extension is appended that
 * describes which table to take it from.
 * 
 * @param tpwd The password to decrypt the key data. */
void Broker5Server::streamPlainTextKey(int threadid, const char *extended_ownername, const char *tpwd)
{
   char *ex_copy = (extended_ownername) ? strdup(extended_ownername) : NULL;
   if (extended_ownername == NULL) return;
   int haspurpose = 0;
   Key::keypurposetype ktype;
   // If the extension matches, set the type to 
   char *isenc = strstr(ex_copy, ".enc");
   if (isenc) { ktype = Key::KEYPURPOSE_ENCRYPTION; haspurpose = 1; isenc[0] = '\0'; }
   char *isdec = strstr(ex_copy, ".dec");
   if (isdec) { ktype = Key::KEYPURPOSE_DECRYPTION; haspurpose = 1; isdec[0] = '\0'; }
   char *issgn = strstr(ex_copy, ".sgn");
   if (issgn) { ktype = Key::KEYPURPOSE_SIGNING; haspurpose = 1; issgn[0] = '\0'; }
   char *ischk = strstr(ex_copy, ".chk");
   if (ischk) { ktype = Key::KEYPURPOSE_CHECKING; haspurpose = 1; ischk[0] = '\0'; }
   if (haspurpose == 0) { if (ex_copy) free(ex_copy); return; }
   const KeyTableView *ktable = NULL;
   switch (ktype)
   {
      case Key::KEYPURPOSE_ENCRYPTION: ktable = dbs->keydb->encryptionTable(); break;
      case Key::KEYPURPOSE_DECRYPTION: ktable = dbs->keydb->decryptionTable(); break;
      case Key::KEYPURPOSE_SIGNING: ktable = dbs->keydb->signingTable(); break;
      case Key::KEYPURPOSE_CHECKING: ktable = dbs->keydb->checkingTable(); break;
   }
   const char *kname = ktable->getLatestKeyNameOfOwner(ex_copy);
   Key t_key(dbs->keydb, kname, ktype);
   const char *tk_name = t_key.getName();
   const char *tk_type = t_key.getKeyType();
   const char *tk_date = t_key.getDateCreated();
   const char *tk_owner = t_key.getOwner();
   char *tk_kdata = NULL;
   AbstractKey *abskey = t_key.getKey((tpwd) ? tpwd : ktable->getWIPPassword());
   abskey->exportKeyToASCII(&tk_kdata);
   static const char *first_header = "HTTP/1.0 200 OK" CRLF_STR CRLF_STR;
   pipearray[threadid].handler_to_send->writeToStream(first_header, strlen(first_header));
   pipearray[threadid].handler_to_send->writefToStream("%s" CRLF_STR, tk_name);
   pipearray[threadid].handler_to_send->writefToStream("%s" CRLF_STR, tk_type);
   pipearray[threadid].handler_to_send->writefToStream("%s" CRLF_STR, tk_date);
   pipearray[threadid].handler_to_send->writefToStream("%s" CRLF_STR, tk_owner);
   if (tk_kdata != NULL)
      pipearray[threadid].handler_to_send->writefToStream(tk_kdata, strlen(tk_kdata));
   pipearray[threadid].handler_to_send->writeEOS();
   delete abskey;
}
